package com.heima;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

@EnableBinding(CountNumListener.CountNumProcessor.class)
public class CountNumListener {

    @StreamListener("count-num-stream-topic")
    @SendTo("count-num-stream-consumer")
    public KStream<String, String> process(KStream<String, String> input) {
        return input
                .flatMap(new KeyValueMapper<String, String, Iterable<KeyValue<String, Integer>>>() {
                    @Override
                    public Iterable<KeyValue<String, Integer>> apply(String key, String value) {
                        String[] split = value.split(":");
                        List<KeyValue<String, Integer>> list = new ArrayList<>();
                        list.add(new KeyValue<>(split[0], Integer.parseInt(split[1])));
                        return list;
                    }
                })
                .groupByKey(Grouped.with(Serdes.String(), Serdes.Integer()))
                .windowedBy(TimeWindows.of(Duration.ofSeconds(15)).grace(Duration.ZERO))
                .aggregate(() -> 0, (key, value, agg) -> value+agg, Materialized.with(Serdes.String(), Serdes.Integer())).suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                .toStream()
                .map((key, value) -> new KeyValue<>(key.key(), String.valueOf(value)));
    }

    public interface CountNumProcessor {
        @Input("count-num-stream-topic")
        KStream<String, String> input();

        @Output("count-num-stream-consumer")
        KStream<String, String> output();
    }
}